package com.rc.saas.mini.config;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * Created by Sven on 2021/02/09
 */
@Component
public class EvchargerKafkaProducer {
    private static Logger logger = LogManager.getLogger(EvchargerKafkaProducer.class);

    @Autowired
    KafkaTemplate kafkaTemplate;//KafkaTemplate是Spring Boot自动配置的，直接使用即可

    //自定义topic
    public static final String TOPIC_TEST = "command1.test";

    //
    public static final String TOPIC_GROUP1 = "topic.group1";

    //
    public static final String TOPIC_GROUP2 = "topic.group2";

    /**
     * 直接发送
     */
    public void send(String topic, Object message) {
        try {
            if (message != null) {
                kafkaTemplate.send(TOPIC_TEST, message);
                logger.info("发送成功:{},:{}", TOPIC_TEST, message);
            }
        } catch (Exception e) {
            logger.error("send 报错", e);
        }

        //kafkaTemplate提供了一个回调方法addCallback，我们可以在回调方法中监控消息是否发送成功 或 失败时做补偿处理
        //kafkaTemplate.send("topic1", message).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
        //    @Override
        //    public void onFailure(Throwable ex) {
        //        System.out.println("发送消息失败：" + ex.getMessage());
        //    }
        //
        //    @Override
        //    public void onSuccess(SendResult<String, Object> result) {
        //        System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
        //                + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
        //    }
        //});
    }
}
